package u.can.i.up.gearmanimpl;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.gearman.Gearman;
import org.gearman.GearmanFunction;
import org.gearman.GearmanFunctionCallback;
import org.gearman.GearmanServer;
import org.gearman.GearmanWorker;

import u.can.i.up.gearmanimpl.common.Configuration;
import u.can.i.up.gearmanimpl.common.Constants;

/**
 * The echo worker polls jobs from a job server and execute the echo function.
 * 
 * The echo worker illustrates how to setup a basic worker
 */
public class Worker implements GearmanFunction {
	/** The echo function name */
	public static final String FUNCTION_NAME = "HTTP";

	/** The host address of the job server */
	public static final String SERVER_HOST =  Configuration.get(Constants.SERVERIP, "127.0.0.1");

	/** The port number the job server is listening on */
	public static final int SERVER_PORT = Configuration.getInt(Constants.SERVERPORT, 4730);

	static Logger logger = LogManager.getLogger(Worker.class);

	public static void main(String... args) {

		/*
		 * Create a Gearman instance
		 */
		Gearman gearman = Gearman.createGearman();

		/*
		 * Create the job server object. This call creates an object represents
		 * a remote job server.
		 * 
		 * Parameter 1: the host address of the job server.
		 * Parameter 2: the port number the job server is listening on.
		 * 
		 * A job server receives jobs from clients and distributes them to
		 * registered workers.
		 */
		GearmanServer server = gearman.createGearmanServer(SERVER_HOST, SERVER_PORT);

		/*
		 * Create a gearman worker. The worker poll jobs from the server and
		 * executes the corresponding GearmanFunction
		 */
		GearmanWorker worker = gearman.createGearmanWorker();

		/*
		 *  Tell the worker how to perform the echo function
		 */
		worker.addFunction(FUNCTION_NAME, new Worker());

		/*
		 *  Tell the worker that it may communicate with the this job server
		 */
		worker.addServer(server);
	}
	/**@author lczgywzyy
	 * @param function - "scheduler"
	 * 					data - byte[]类型，会默认转成String类型，为Ftp上的绝对路径
	 * */
	public byte[] work(String function, byte[] data, GearmanFunctionCallback callback){
		sendMessage();
		logger.info("WORKERRRRRRRRRRR!!!!!!!!!");
		return "".toString().getBytes();  
	}

	public void sendMessage() {
		HttpURLConnection urlConnection = null;
		HashMap<String, String> hashParam = new HashMap<String, String>();
		String url = "http://192.168.2.138:6543/CeleryPost";
		hashParam.put("a", "b");

		int type = 1;
		switch (type) {
		case 1:
			initConnectionPost(url, urlConnection, hashParam);
			break;
		default:
			return;
		}

	}

	private void initConnection(HttpURLConnection urlConnection){
		urlConnection.setDoInput(true);
		urlConnection.setDoOutput(true);
		urlConnection.setRequestProperty("connection", "Keep-Alive");
		urlConnection.setRequestProperty("Accept-Charset", "utf-8");
		urlConnection.setConnectTimeout(3000);
	}

	private void initConnectionPost(String url, HttpURLConnection urlConnection, HashMap<String, String> hashParam) {
		try {
			urlConnection = (HttpURLConnection) new URL(url).openConnection();
			urlConnection.setRequestMethod("POST");
			initConnection(urlConnection);
			initParamsPost(urlConnection, hashParam);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * 提交表单*
	 */
	@SuppressWarnings("rawtypes")
	private void initParamsPost(HttpURLConnection urlConnection, HashMap<String, String> hashParam) {
		urlConnection.setRequestProperty("Content-Type", "application/x-www-form-urlencoded");
		StringBuilder buffer = new StringBuilder();
		Iterator iterator = hashParam.entrySet().iterator();
		while (iterator.hasNext()) {
			Map.Entry entry = (Map.Entry) iterator.next();
			String key = String.valueOf(entry.getKey());
			String val = String.valueOf(entry.getValue());
			buffer.append(key);
			buffer.append("=");
			buffer.append(val);
			if (iterator.hasNext()) {
				buffer.append("&");
			}
		}
		try {
			urlConnection.getOutputStream().write(buffer.toString().getBytes());
			int responseCode = urlConnection.getResponseCode();  
			if (responseCode == 200) {  
				// 获得输入流，从服务器端获得数据  
				InputStream inputStream = (InputStream) urlConnection.getInputStream();  
				String result = changeInputStream(inputStream, "utf-8");
				logger.info(result);
			}  
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/* 
	 * // 把从输入流InputStream按指定编码格式encode变成字符串String 
	 */  
	public static String changeInputStream(InputStream inputStream, String encode) {  
		// ByteArrayOutputStream 一般叫做内存流  
		ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();  
		byte[] data = new byte[1024];  
		int len = 0;  
		String result = "";  
		if (inputStream != null) {  
			try {  
				while ((len = inputStream.read(data)) != -1) {  
					byteArrayOutputStream.write(data, 0, len);  
				}  
				result = new String(byteArrayOutputStream.toByteArray(), encode);  
			} catch (IOException e) {  
				// TODO Auto-generated catch block  
				e.printStackTrace();  
			}  
		}  
		return result;  
	}  

}